package com.cuit.remoting.transport.netty.client;

import com.cuit.enums.RpcErrorMessageEnum;
import com.cuit.enums.RpcResponseCodeEnum;
import com.cuit.exception.RpcException;
import com.cuit.extension.ExtensionLoader;
import com.cuit.factory.SingleFactory;
import com.cuit.filter.FilterContext;
import com.cuit.filter.FilterChain;
import com.cuit.filter.FilterChainController;
import com.cuit.job.JobManager;
import com.cuit.mock.MockServiceManager;
import com.cuit.registry.ServiceDiscovery;
import com.cuit.remoting.dto.RpcRequest;
import com.cuit.remoting.dto.RpcResponse;
import com.cuit.remoting.transport.RpcRequestTransport;
import com.cuit.remoting.transport.netty.Codec.RpcMessageDecoder;
import com.cuit.remoting.transport.netty.Codec.RpcMessageEncoder;
import com.cuit.retryies.ExponetialBackOffRetry;
import com.cuit.retryies.RetryPolicy;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;



import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.concurrent.*;

/**
 * @PROJECT_NAME: rpc
 * @SCHOOL:CUIT
 * @USER: Liquid
 * @DATE: 2022/2/23 16:26
 */
@Slf4j
public class NettyRpcClient implements RpcRequestTransport {
    private final ChannelProvider channelProvider;
    private final FilterChainController filterChainController;
    private Bootstrap bootstrap;
    private EventLoopGroup group;;
    private final ServiceDiscovery serviceDiscovery;
    private final JobManager jobManager;
    private final MockServiceManager mockServiceManager;
    private final HashedWheelTimer timer;
    private final ExecutorService executorService;

    private final RetryPolicy retryPolicy;

    public NettyRpcClient(){
        channelProvider = SingleFactory.getSingle(ChannelProvider.class);
        serviceDiscovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class).getExtension("zk");
        filterChainController = SingleFactory.getSingle(FilterChainController.class);
        retryPolicy = new ExponetialBackOffRetry(1000, Integer.MAX_VALUE);
        jobManager = SingleFactory.getSingle(JobManager.class);
        timer = SingleFactory.getSingle(HashedWheelTimer.class);
        executorService = new ThreadPoolExecutor(10, 50
                , 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        mockServiceManager = SingleFactory.getSingle(MockServiceManager.class);
    }

    public Channel connect(InetSocketAddress inetSocketAddress){
        if (bootstrap == null){
            group = new NioEventLoopGroup(10, executorService);
            bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline()
                                    .addLast(new IdleStateHandler(0,4,0, TimeUnit.SECONDS))
                                    .addLast(new ConnectionWatchHandler(bootstrap, timer, retryPolicy))
                                    .addLast(new RpcMessageEncoder())
                                    .addLast(new RpcMessageDecoder())
                                    .addLast(new NettyRpcClientHandler())
                            ;
                        }
                    });
        }

        Channel channel = getChannel(inetSocketAddress);
        return channel;
    }
    @Override
    public Object sendRPCRequest(RpcRequest rpcRequest) throws Exception{

        InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest);
        connect(inetSocketAddress);

        CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();
        //过滤链条
        FilterChain consumerFilterChain = filterChainController.getConsumerFilterChain();
        FilterContext context = FilterContext.builder()
                .rpcRequest(rpcRequest)
                .shouldBreak(false)
                .resultFuture(resultFuture)
                .filterChain(consumerFilterChain)
                .build();
        consumerFilterChain.doFilter(context);

        return resultFuture;
    }


    @Override
    public RpcResponse invokeMock(RpcRequest rpcRequest) {
        String rpcServiceName = rpcRequest.getRpcServiceName();
        Object mockService = mockServiceManager.getMockService(rpcServiceName);
        if (mockService == null){
            log.error("客户端没有对远程服务[{}]设置对应的降级服务", rpcServiceName);
        }

        Object data = null;
        try{
            Method method = mockService.getClass().getMethod(rpcRequest.getMethodName(), rpcRequest.getParameterTypes());
            log.info("发送服务降级，客户端调用远程服务[{}]对应的本地的降级服务", rpcServiceName);
            data = method.invoke(mockService, rpcRequest.getParameters());
        }catch (InvocationTargetException e) {
            throw  new RpcException(RpcErrorMessageEnum.SERVICE_MOCK_FAILED);
        }catch (Exception e){
            throw new RpcException(e.getMessage(), e);
        }
        RpcResponse<Object> rpcResponse = RpcResponse.builder()
                .code(RpcResponseCodeEnum.SUCCESS.getCode()).requestId(rpcRequest.getRequestId()).data(data).build();
        return rpcResponse;
    }

    @SneakyThrows
    public Channel doConnect(InetSocketAddress inetSocketAddress) {
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("The client has connected [{}] successful!", inetSocketAddress.toString());
                completableFuture.complete(future.channel());
            }
        });
        return completableFuture.get();
    }

    public Channel getChannel(InetSocketAddress inetSocketAddress) {
        Channel channel = channelProvider.get(inetSocketAddress);
        if (channel == null) {
            channel = doConnect(inetSocketAddress);
            channelProvider.set(inetSocketAddress, channel);
        }
        return channel;
    }

    public void close(){
        group.shutdownGracefully();
    }


}
